package com.openapi.service;

import com.openapi.config.JT808ServerConfig;
import exsun.jt808.data.base.MsgDirection;
import exsun.jt808.data.kafka_data.KafkaJT808Data;
import exsun.jt808.data.upload_data.Jt808EntityBase;
import exsun.jt808.data.util.GsonUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class KafkaService {

	@Resource(name = "HandleCenterKafkaTemplate")
	private KafkaTemplate<String, String> stringkafkaTemplate;
	@Autowired
	JT808ServerConfig jt808ServerConfig;

	public void push(String topic,String key, String info) {
		this.stringkafkaTemplate.send(topic, key, info);
	}

	public void push(Jt808EntityBase data)
	{
		int port = jt808ServerConfig.getJt808ServerPort();
		KafkaJT808Data kafkaJT808Data = new KafkaJT808Data(data,data.dvo,data.mid,data.tpc, MsgDirection.up, port);
		String sendData = kafkaJT808Data.getJsonFromObject();
		stringkafkaTemplate.send(data.tpc, kafkaJT808Data.getDevNo(), sendData);
		//logger.info("kafka 推送信息----" + data.getTpc() + "--解析时间--" + data.getPte() + "--设备号--" + data.dvo);
	}

	public void push(String devNo, String topic, Object data)
	{
		int port = jt808ServerConfig.getJt808ServerPort();
		KafkaJT808Data kafkaJT808Data = new KafkaJT808Data(data, devNo,0, topic, MsgDirection.up,port);

		String sendData = kafkaJT808Data.getJsonFromObject();

		stringkafkaTemplate.send(topic, kafkaJT808Data.getDevNo(), sendData);
		//logger.info("kafka 推送信息----" + data.getTpc() + "--解析时间--" + data.getPte() + "--设备号--" + data.dvo);
	}

	public void pushGeneralData(String devNo, String topic, Object data)
	{

		String sendData = GsonUtils.getJsonFromObject(data);

		stringkafkaTemplate.send(topic, devNo, sendData);
		//logger.info("kafka 推送信息----" + data.getTpc() + "--解析时间--" + data.getPte() + "--设备号--" + data.dvo);
	}

}